/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_infra.cpp
 * Interface for HTAP entry point Facade.
 *
 * IDENTIFICATION
 * src/gausskernel/storage/htap/htap_infra.cpp
 *
 * -------------------------------------------------------------------------
 */

#include "htap_common.h"
#include "htap_infra.h"
#include "htap_helpers.h"
#include "htap_debug.h"
#include "htap_statistics.h"
#include "config_manager.h"
#include "statistics_manager.h"
#include "cycles.h"
#include "htap_flush.h"
#include "htap_internal.h"
#include "htap_xlog_redoer.h"
#include "mot_engine.h"

namespace htap {
HTAP_IMPLEMENT_CLASS_LOGGER(HTAPInfra, HTAP)

#define raiseAbortTxnError()                                                            \
    {                                                                                   \
        ereport(ERROR, (errmodule(MOD_MOT), errcode(ERRCODE_IN_FAILED_SQL_TRANSACTION), \
            errmsg("current transaction is aborted, "                                   \
            "commands ignored until end of transaction block, firstChar[%c]",           \
            u_sess->proc_cxt.firstChar),                                                \
            0));                                                                        \
    }

HTAPInfra *HTAPInfra::sInstance = nullptr;

inline bool IsIMCStoredTable(Oid relationId)
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    return (txn->GetTableByExternalId(relationId) != nullptr);
}

static void MemoryEreportError()
{
    int result = MOT::GetLastError();
    if (result == MOT_ERROR_INVALID_MEMORY_SIZE) {
        ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("Invalid memory allocation request size.")));
    } else if (result == MOT_ERROR_OOM) {
        ereport(ERROR, (errcode(ERRCODE_OUT_OF_LOGICAL_MEMORY), errmsg("Memory is temporarily unavailable.")));
    }
}

/**
 * Destroys scan cursors.
 * @param txn MOT transaction manager
 */
inline void CleanQueryStatesOnError(MOT::TxnManager *txn)
{
    if (txn != nullptr) {
        for (auto &itr : txn->m_queryState) {
            MOT::IndexIterator *cursor = (MOT::IndexIterator *)itr.second;
            if (cursor != nullptr) {
                cursor->Invalidate();
                cursor->Destroy();
                delete cursor;
            }
        }
        txn->m_queryState.clear();
    }
}

bool HTAPInfra::HTAPTxnEnsure(bool registerCallbacks /* = true */)
{
    HTAPInfra *htap = HTAPInfra::GetInstance();
    if (htap != nullptr) {
        if (!HTAPEnsureSafeThreadAccess()) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Internal error", "Failed to set up thread local variables");
            return false;
        }
        if (HTAPGetSafeTxn() == nullptr) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Internal error", "Failed to set up current transaction manager");
            return false;
        }
        if (registerCallbacks) {
            htap->EnsureTxnCallbacksRegistered();
        }
        return true;
    }
    return false;
}

bool HTAPInfra::CreateInstance(const char *configFilePath)
{
    if (sInstance == nullptr) {
        sInstance = new (std::nothrow) HTAPInfra();
        if (sInstance == nullptr) {
            HTAP_REPORT_ERROR(HTAP_ERROR_OOM, "Init HTAP", "Failed to create HTAPInfra");
            return false;
        }
        if (!sInstance->Initialize(configFilePath)) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to initialize HTAPInfra");
            HTAPInfra::DestroyInstance();
            return false;
        }
        if (!sInstance->Start()) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to start HTAPInfra");
            HTAPInfra::DestroyInstance();
            return false;
        }
    }
    return true;
}

void HTAPInfra::DestroyInstance()
{
    if (sInstance != nullptr) {
        delete sInstance;
        sInstance = nullptr;
    }
}

HTAPInfra *HTAPInfra::GetInstance()
{
    return sInstance;
}

bool HTAPInfra::Initialize(const char *configFilePath)
{
    HTAP_LOG_INFO("Initializing HTAP Infra Facade");

    // Create and initialize all core services (configuration, statistics, etc.)
    if (!InitializeCoreServices(configFilePath)) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to initialize core services");
        return false;
    }

    // Create and initialize all sub-components (applicative services)
    if (!InitializeApplicativeServices()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to initialize applicative services");
        return false;
    }

    HTAP_LOG_INFO("HTAP Infra initialized successfully");
    return true;
}

void HTAPInfra::Terminate()
{
    HTAP_LOG_INFO("Terminating HTAP Infra Facade");

    // first stop all services
    StopAllServices();

    // terminate all sub-components (applicative services)
    // destroy flush
    HTAPFlush::DestroyInstance();
    m_flush = nullptr;
    
    // destroy flush
    HTAPXLogRedoer::DestroyInstance();
    m_xLogRedoer = nullptr;

    // terminate all core services (configuration, statistics, etc.)
    if (m_instanceMemoryContext != nullptr) {
        MemoryContextDelete(m_instanceMemoryContext);
        m_instanceMemoryContext = nullptr;
    }
    FlushStatisticsProvider::DestroyInstance();
    DeltaStatisticsProvider::DestroyInstance();
    ModifyStatisticsProvider::DestroyInstance();

    HTAP_LOG_INFO("HTAP Infra terminated successfully");
}

bool HTAPInfra::Start()
{
    HTAP_LOG_INFO("Starting HTAP Facade");
    if (!StartStatistics()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to start statistics service");
        return false;
    }
    HTAP_LOG_INFO("HTAP started successfully");
    return true;
}

void HTAPInfra::Stop()
{
    HTAP_LOG_INFO("Stopping HTAP Facade");
    StopAllServices();
    HTAP_LOG_INFO("HTAP stopped successfully");
}

bool HTAPInfra::InitializeCoreServices(const char *configFilePath)
{
    HTAP_LOG_INFO("Initializing HTAP core services");

    // create configuration
    // print the canonical path of the configuration file
    if (configFilePath == nullptr) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INVALID_ARG, "Load HTAP Configuration",
                          "Cannot initialization configuration: missing configuration file path");
        return false;
    }

    if (configFilePath != nullptr) {
        char *resolvedPath = (char *)malloc(sizeof(char) * PATH_MAX);
        if (resolvedPath == nullptr) {
            HTAP_LOG_ERROR("Failed to allocate %u bytes for configuration loading",
                           (unsigned)(sizeof(char) * PATH_MAX));
        } else {
            if (!realpath(configFilePath, resolvedPath)) {
                HTAP_LOG_SYSTEM_WARN(realpath, "Failed to get resolved path from path %s", configFilePath);
                errno_t erc = strcpy_s(resolvedPath, PATH_MAX, "N/A");
                securec_check(erc, "\0", "\0");
            }
            HTAP_LOG_INFO("Startup: Loading configuration from %s", resolvedPath);
            free(resolvedPath);
        }
    }

    // register global configuration as first listener (other listeners rely on it)
    if (!MOT::ConfigManager::GetInstance().AddConfigChangeListener(&m_cfg)) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Load HTAP Configuration",
                          "Failed to register main configuration listener at the configuration manager");
        return false;
    }

    // add configuration file loader
    if (!MOT::ConfigManager::GetInstance().AddConfigFile(configFilePath, "Main",
                                                         MOT::ConfigFileFormat::CONFIG_FILE_PROPS)) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Load HTAP Configuration",
                          "Failed to add configuration file %s to configuration manager", configFilePath);
        return false;
    }

    // trigger reloading of all configuration trees
    if (!MOT::ConfigManager::GetInstance().InitLoad()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Load HTAP Configuration", "Failed to reload configuration manager");
        return false;
    }

    // now update all configuration from file
    m_cfg.LoadConfig();
    if (m_cfg.m_logFormat == MOT::LOG_FORMAT_INTERNAL) {
        HTAP_LOG_INFO("Switching to internal log format");
        MOT::SetLogFormatInternal("HTAP");
    } else if (m_cfg.m_logFormat == MOT::LOG_FORMAT_EXTERNAL) {
        HTAP_LOG_INFO("Switching to external log format");
        MOT::SetLogFormatExternal(MOD_HTAP);
    }

    HTAP_LOG_INFO("Startup: HTAP configuration initialized");

    // create statistics
    if (!ModifyStatisticsProvider::CreateInstance()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Init", "Failed to create HTAP hook statistics provider");
        return false;
    }
    if (!DeltaStatisticsProvider::CreateInstance()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Init", "Failed to create HTAP delta statistics provider");
        return false;
    }
    if (!FlushStatisticsProvider::CreateInstance()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Init", "Failed to create HTAP flush statistics provider");
        return false;
    }

    // initialize instance-level memory controller
    m_instanceMemoryContext = CreateMemoryContext(nullptr, "HTAP memory context");
    if (m_instanceMemoryContext == nullptr) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to create top memory context");
        return false;
    }

    HTAP_LOG_INFO("Core services initialized successfully");
    return true;
}

bool HTAPInfra::InitializeApplicativeServices()
{
    HTAP_LOG_INFO("Initializing HTAP applicative services");

    // create the HTAP Flush
    if (!htap::HTAPFlush::CreateInstance()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to create HTAP flush");
        return false;
    }
    m_flush = HTAPFlush::GetInstance();

    // create the HTAP xlog redoer
    if (!htap::HTAPXLogRedoer::CreateInstance()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to create HTAP xlog redoer");
        return false;
    }
    m_xLogRedoer = HTAPXLogRedoer::GetInstance();

    return true;
}

void HTAPInfra::StopAllServices()
{
    HTAP_LOG_INFO("Stopping HTAP applicative services");
    if ((m_flush != nullptr) && m_flush->IsRunning()) {
        HTAP_LOG_INFO("Stopping HTAP flush");
        if (m_flush->Stop() != HTAP_RC_OK) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Term HTAP", "Failed to stop flush thread");
        }
    }
    if ((m_xLogRedoer != nullptr) && m_xLogRedoer->IsRunning()) {
        HTAP_LOG_INFO("Stopping HTAP xlog redoer");
        if (m_xLogRedoer->Stop() != HTAP_RC_OK) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Term HTAP", "Failed to stop xlog redoer thread");
        }
    }
    HTAP_LOG_INFO("HTAP applicative services stopped");

    // stop core services, now we have only statistics report thread
    HTAP_LOG_INFO("Stopping HTAP core services");
    if (m_shouldStopStatistics) {
        HTAP_LOG_INFO("Stopping statistics reporting thread");
        MOT::StatisticsManager::GetInstance().Stop();
    }
    HTAP_LOG_INFO("HTAP core services stopped");
}

bool HTAPInfra::StartStatistics()
{
    HTAP_LOG_INFO("Starting HTAP statistics");
    bool shouldStartStatistics = false;
#ifdef ENABLE_MOT
    if (!MOT::GetGlobalConfiguration().m_enableStats && m_cfg.m_enableStats) {
        shouldStartStatistics = true;
    }
#else
    shouldStartStatistics = m_cfg.m_enableStats;
#endif
    if (shouldStartStatistics) {
        HTAP_LOG_INFO("Launching HTAP statistics reporting thread");
        if (!MOT::StatisticsManager::GetInstance().Start()) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Init", "Failed to start statistics manager");
            return false;
        }
        m_shouldStopStatistics = true;
    }
    HTAP_LOG_INFO("HTAP statistics started successfully");
    return true;
}

ThreadId HTAPInfra::StartFlush()
{
    ThreadId threadId = 0;
    if (m_flush != nullptr) {
        if (m_flush->IsRunning()) {
            HTAP_LOG_WARN("Request to start flush thread rejected: flush thread is already running");
            threadId = m_flush->GetThreadId();
        } else {
            RC res = m_flush->Start(threadId);
            if (res != HTAP_RC_OK) {
                HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Start Flush", "Failed to start flush thread");
            }
        }
    }
    return threadId;
}

ThreadId HTAPInfra::StartXLogRedoer()
{
    ThreadId threadId = 0;
    if (m_xLogRedoer != nullptr) {
        if (m_xLogRedoer->IsRunning()) {
            HTAP_LOG_WARN("Request to start htap xlog redoer thread rejected: htap xlog redoer already running");
            threadId = m_xLogRedoer->GetThreadId();
        } else {
            RC res = m_xLogRedoer->Start(threadId);
            if (res != HTAP_RC_OK) {
                HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Start XLogRedoer",
                    "Failed to start htap xlog redoer thread");
            }
        }
    }
    return threadId;
}

MemoryContext HTAPInfra::GetThreadMemoryContext()
{
    HTAP_ASSERT(m_instanceMemoryContext != nullptr);
    if (t_thrd.top_mem_cxt == nullptr) {
        t_thrd.top_mem_cxt = CreateMemoryContext(m_instanceMemoryContext, "HTAP thread memory");
        t_thrd.htap_cxt.memory_context_owner = true;
    }
    return t_thrd.top_mem_cxt;
}

void HTAPInfra::DestroyThreadMemoryContext()
{
    if (t_thrd.htap_cxt.memory_context_owner && (t_thrd.top_mem_cxt != nullptr)) {
        DestroyMemoryContext(t_thrd.top_mem_cxt);
        t_thrd.top_mem_cxt = nullptr;
        t_thrd.htap_cxt.memory_context_owner = false;
    }
}

void HTAPInfra::DestroySessionMemoryContext()
{
    if ((u_sess != nullptr) && u_sess->htap_cxt.memory_context_owner && (u_sess->top_mem_cxt != nullptr)) {
        DestroyMemoryContext(u_sess->top_mem_cxt);
        u_sess->top_mem_cxt = nullptr;
        u_sess->htap_cxt.memory_context_owner = false;
    }
}

MemoryContext HTAPInfra::CreateMemoryContext(MemoryContext parent, const char *name)
{
    MemoryContext result = nullptr;
    PG_TRY();
    {
        result = AllocSetContextCreate(parent, name, ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE,
            ALLOCSET_DEFAULT_MAXSIZE, STANDARD_CONTEXT);
    }
    PG_CATCH();
    {
        HTAP_LOG_ERROR_STACK("Failed to create memory context %s: %s (%d)", name, Geterrmsg(), geterrcode());
        FlushErrorState();
    }
    PG_END_TRY();
    return result;
}

void HTAPInfra::DestroyMemoryContext(MemoryContext memoryContext)
{
    if (memoryContext != nullptr) {
        MemoryContextDelete(memoryContext);
    }
}

RC HTAPInfra::OnInsertRow(Oid relationId, ItemPointer ctid, TupleTableSlot *slot)
{
    // determine first whether this is an imcstored relation, then call pre-hook
    if (!IsIMCStoredTable(relationId)) {
        HTAP_LOG_DIAG1("Skipping hook of insert-row call of non-HTAP table with relation id %u", relationId);
        return HTAP_RC_EVENT_DISCARDED;
    }

    // execute required pre-hook actions
    if (unlikely(!checkBeforeModify())) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Insert Row Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    // handle event
    HTAP_LOG_DEBUG("Hooking insert-row call with relation id %u and CTID 0x%" PRIx64, relationId,
        HTAPAdaptor::GetCtid(ctid));
    return HandleInsertRow(relationId, ctid, slot);
}

RC HTAPInfra::OnUpdateRow(Oid relationId, ItemPointer oldCtid, ItemPointer newCtid, TupleTableSlot *slot)
{
    // determine first whether this is an imcstored relation, then call pre-hook
    if (!IsIMCStoredTable(relationId)) {
        HTAP_LOG_DIAG1("Skipping hook of update-row call of non-HTAP table with relation id %u", relationId);
        return HTAP_RC_EVENT_DISCARDED;
    }

    // execute required pre-hook actions
    if (unlikely(!checkBeforeModify())) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Update Row Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    HTAP_LOG_DEBUG("Hooking update-row call with relation id %u, old CTID 0x%" PRIx64 " and new CTID %" PRIx64,
        relationId, HTAPAdaptor::GetCtid(oldCtid), HTAPAdaptor::GetCtid(newCtid));

    // handle event
    RC result = HTAP_RC_OK;
    if (ItemPointerEquals(oldCtid, newCtid)) {
        HTAP_LOG_DEBUG("Updating in-place tuple with CTID %" PRIx64, HTAPAdaptor::GetCtid(oldCtid));
        result = HandleUpdateRow(relationId, oldCtid, slot);
    } else {
        HTAP_LOG_DEBUG("Updated tuple CTID 0x%" PRIx64 " changed to 0x%" PRIx64, HTAPAdaptor::GetCtid(oldCtid),
            HTAPAdaptor::GetCtid(newCtid));
        result = HandleDeleteRow(relationId, oldCtid, true);
        if (result == HTAP_RC_OK) {
            result = HandleInsertRow(relationId, newCtid, slot);
        }
    }
    return result;
}

RC HTAPInfra::OnDeleteRow(Oid relationId, ItemPointer ctid)
{
    // determine first whether this is an imcstored relation, then call pre-hook
    if (!IsIMCStoredTable(relationId)) {
        HTAP_LOG_DIAG1("Skipping hook of delete-row call of non-HTAP table with relation id %u", relationId);
        return HTAP_RC_EVENT_DISCARDED;
    }

    // execute required pre-hook actions
    if (unlikely(!checkBeforeModify())) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Delete Row Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    // handle event
    HTAP_LOG_DEBUG("Hooking delete-row call with relation id %u and CTID 0x%" PRIx64, relationId,
        HTAPAdaptor::GetCtid(ctid));
    return HandleDeleteRow(relationId, ctid, false);
}

RC HTAPInfra::OnCreateTable(Oid parentID, Oid relationId)
{
    HTAP_LOG_DEBUG("Hooking create imcstored table call with relation id %u and parent id %u", relationId, parentID);
    if (!checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Create Table Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    RC result = HandleCreateTable(parentID, relationId);
    if (result != HTAP_RC_OK) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Create Table Hook", "Failed to create HTAP delta table");
        return result;
    }
    return HTAP_RC_OK;
}

RC HTAPInfra::OnDropTable(Oid relationId, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    HTAP_LOG_DEBUG("Hooking drop imcstored table call with relation id %u", relationId);
    if (!checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Drop Table Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    RC result = HandleDropTable(relationId, out_oid_del_info);
    if (result != HTAP_RC_OK && result != HTAP_RC_TABLE_NOT_FOUND) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Drop Table Hook", "Failed to drop HTAP delta table");
        return result;
    }
    return HTAP_RC_OK;
}

RC HTAPInfra::OnTruncateTable(Oid relationId, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    HTAP_LOG_DEBUG("Hooking truncate imcstored table call with relation id %u", relationId);
    if (!checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Truncate Table Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    RC result = HandleTruncateTable(relationId, out_oid_del_info);
    if (result != HTAP_RC_OK) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Truncate Table Hook", "Failed to truncate HTAP delta table");
        return result;
    }
    return HTAP_RC_OK;
}

RC HTAPInfra::OnSearchDelta(Oid relationId, uint64_t &rowCount, uint64_t &newRowCount, uint64_t &memSize)
{
    HTAP_LOG_DEBUG("Hooking Query Deltae of imcstored table call with relation id %u", relationId);
    if (!checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Query Delta Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    RC result = HandleSearchDelta(relationId, rowCount, newRowCount, memSize);

    if (result != HTAP_RC_OK) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Query Delta Hook", "Failed to query HTAP delta table");
        return result;
    }

    return HTAP_RC_OK;
}

RC HTAPInfra::OnAlterTable(Oid relationId, Relation rel, AlterTableStmt *stmt)
{
    HTAP_LOG_DEBUG("Hooking alter table call with relation id %u", relationId);
    if (HTAP_CHECK_DEBUG_LOG_LEVEL()) {
        char *parsedStmt = nodeToString(stmt);
        HTAP_LOG_DEBUG("Checking alter table statement: %s", parsedStmt);
        pfree(parsedStmt);
    }

    if (!checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Alter Table Hook", "Failed to execute pre-hook actions");
        return GetRootError();
    }

    RC result = HandleAlterTable(relationId, rel, stmt);
    if (result != HTAP_RC_OK) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Alter Table Hook", "Failed to drop HTAP delta table");
        return result;
    }
    return HTAP_RC_OK;
}

void HTAPInfra::HTAPXactCallback(XactEvent event, void *arg)
{
    htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
    if (unlikely(htap == nullptr)) {
        HTAP_LOG_TRACE("HTAPInfra is null, %s hook discarded", "HTAPXactCallback");
        return;
    }

    if (!htap->checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAPXactCallback", "Failed to execute pre-hook actions");
        HTAP_LOG_ERROR_EX("Failed to execute %s hook: %s", "HTAPXactCallback", htap::RcToString(htap::GetRootError()));
        return;
    }
    (void)htap->HandleXactEvent(event);
}

void HTAPInfra::HTAPSubxactCallback(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid,
    void *arg)
{
    htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
    if (unlikely(htap == nullptr)) {
        HTAP_LOG_TRACE("HTAPInfra is null, %s hook discarded", "HTAPSubxactCallback");
        return;
    }

    if (!htap->checkBeforeModify()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAPSubxactCallback", "Failed to execute pre-hook actions");
        HTAP_LOG_ERROR_EX("Failed to execute %s hook: %s", "HTAPSubxactCallback",
            htap::RcToString(htap::GetRootError()));
        return;
    }
    (void)htap->HandleSubXactEvent(event, mySubid, parentSubid);
}

RC HTAPInfra::HandleInsertRow(Oid relationId, ItemPointer ctid, TupleTableSlot *slot)
{
    return htap::HTAPAdaptor::TableInsert(relationId, ctid, slot);
}

RC HTAPInfra::HandleUpdateRow(Oid relationId, ItemPointer ctid, TupleTableSlot *slot)
{
    return htap::HTAPAdaptor::TableModify(relationId, ctid, slot);
}

RC HTAPInfra::HandleDeleteRow(Oid relationId, ItemPointer ctid, bool fromUpd)
{
    return htap::HTAPAdaptor::TableDelete(relationId, ctid, fromUpd);
}

RC HTAPInfra::HandleCreateTable(Oid parentID, Oid relationId)
{
    return htap::HTAPAdaptor::CreateTable(parentID, relationId);
}

RC HTAPInfra::HandleDropTable(Oid relationId, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    return htap::HTAPAdaptor::DropTable(relationId, out_oid_del_info);
}

RC HTAPInfra::HandleSearchDelta(Oid relationId, uint64_t &rowCount, uint64_t &newRowCount, uint64_t &memSize)
{
    return htap::HTAPAdaptor::QueryDelta(relationId, rowCount, newRowCount, memSize);
}

RC HTAPInfra::HandleTruncateTable(Oid relationId, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    return htap::HTAPAdaptor::TruncateTable(relationId, out_oid_del_info);
}

RC HTAPInfra::HandleAlterTable(Oid relationId, Relation rel, AlterTableStmt *stmt)
{
    return HTAP_RC_OK;
}

bool HTAPInfra::checkBeforeModify()
{
    ClearErrorStack();
    if (!HTAPEnsureSafeThreadAccess()) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Internal error", "Failed to set up thread local variables");
        return false;
    }
    if (HTAPGetSafeTxn() == nullptr) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Internal error", "Failed to set up current transaction manager");
        return false;
    }
    EnsureTxnCallbacksRegistered();
    return true;
}

RC HTAPInfra::HandleXactEvent(XactEvent event)
{
    MOT::RC rc = MOT::RC_OK;
    MOT::TxnManager *txn = nullptr;
    int saveInterruptHoldoffCount = t_thrd.int_cxt.InterruptHoldoffCount;

    PG_TRY();
    {
        txn = HTAPGetSafeTxn();
    }
    PG_CATCH();
    {
        /*
         * handle ereport error will reset InterruptHoldoffCount issue,
         * if not handle, caller may fail on assert
         */
        t_thrd.int_cxt.InterruptHoldoffCount = saveInterruptHoldoffCount;
        switch (event) {
            case XACT_EVENT_ABORT:
            case XACT_EVENT_ROLLBACK_PREPARED:
            case XACT_EVENT_PREROLLBACK_CLEANUP:
                return HTAP_RC_OK;
            default:
                PG_RE_THROW();
        }
    }
    PG_END_TRY();

    ::TransactionId tid = GetCurrentTransactionIdIfAny();
    if (TransactionIdIsValid(tid)) {
        txn->SetTransactionId(tid);
    }

    MOT::TxnState txnState = txn->GetTxnState();

    if (event == XACT_EVENT_STMT_FINISH) {
        txn->FinishStatement();
    } else if (event == XACT_EVENT_START) {
        if (txnState == MOT::TxnState::TXN_START) {
            // Double start!!!
            txn->Rollback();
        }
        if (txnState != MOT::TxnState::TXN_PREPARE) {
            txn->StartTransaction(tid, u_sess->utils_cxt.XactIsoLevel);
        }
    } else if (event == XACT_EVENT_COMMIT) {
        if (txnState == MOT::TxnState::TXN_END_TRANSACTION) {
            return HTAP_RC_OK;
        }

        if (txnState == MOT::TxnState::TXN_COMMIT) {
            return HTAP_RC_OK;
        }

        if (txnState == MOT::TxnState::TXN_PREPARE) {
            // Transaction is in prepare state, it's 2pc transaction. Nothing to do in XACT_EVENT_COMMIT.
            // Actual CommitPrepared will be done when XACT_EVENT_RECORD_COMMIT is called by the envelope.
            return HTAP_RC_OK;
        }

        if (!IsTransactionBlock() && txn->IsTxnAborted()) {
            elog(DEBUG2, "Implicit transaction aborted by sub-transaction");
            txn->SetTxnState(MOT::TxnState::TXN_ROLLBACK);
            return HTAP_RC_OK;
        }

        if (txn->IsTxnAborted()) {
            raiseAbortTxnError();
        }

        rc = htap::HTAPAdaptor::ValidateCommit();
        if (rc != MOT::RC_OK) {
            MemoryEreportError();
            ereport(ERROR, (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                errmsg("could not serialize access due to concurrent update")));
        }
        txn->SetTxnState(MOT::TxnState::TXN_COMMIT);
    } else if (event == XACT_EVENT_RECORD_COMMIT) {
        if (txnState == MOT::TxnState::TXN_END_TRANSACTION) {
            return HTAP_RC_OK;
        }

        if (txn->IsTxnAborted()) {
            return HTAP_RC_OK;
        }

        MOT_ASSERT(txnState == MOT::TxnState::TXN_COMMIT || txnState == MOT::TxnState::TXN_PREPARE);

        if (htap::HTAPAdaptor::IsTxnWriteSetEmpty()) {
            return HTAP_RC_OK;
        }

        // Need to get the envelope CSN for cross transaction support.
        uint64_t csn = MOT::GetCSNManager().GetNextCSN();
        if (txnState == MOT::TxnState::TXN_PREPARE) {
            htap::HTAPAdaptor::CommitPrepared(csn);
        } else {
            htap::HTAPAdaptor::RecordCommit(csn);
        }
    } else if (event == XACT_EVENT_END_TRANSACTION) {
        if (txnState == MOT::TxnState::TXN_END_TRANSACTION) {
            return HTAP_RC_OK;
        }
        htap::HTAPAdaptor::EndTransaction();
        txn->SetTxnState(MOT::TxnState::TXN_END_TRANSACTION);
    } else if (event == XACT_EVENT_PREPARE) {
        rc = htap::HTAPAdaptor::Prepare();
        if (rc != MOT::RC_OK) {
            elog(DEBUG2, "Prepare failed, abort parent transaction from MOT prepare, tid %lu", tid);
            MemoryEreportError();
            ereport(ERROR, (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                errmsg("could not serialize access due to concurrent update")));
        }
        txn->SetTxnState(MOT::TxnState::TXN_PREPARE);
    } else if (event == XACT_EVENT_ABORT) {
        uint64_t csn = MOT::GetCSNManager().GetGcEpoch();
        txn->SetCommitSequenceNumber(csn);
        htap::HTAPAdaptor::Rollback();
        txn->SetTxnState(MOT::TxnState::TXN_ROLLBACK);
    } else if (event == XACT_EVENT_COMMIT_PREPARED) {
        if (txnState == MOT::TxnState::TXN_PREPARE) {
            // Need to get the envelope CSN for cross transaction support.
            uint64_t csn = MOT::GetCSNManager().GetNextCSN();
            htap::HTAPAdaptor::CommitPrepared(csn);
        } else if (txnState == MOT::TxnState::TXN_START) {
            // Need to get the envelope CSN for cross transaction support.
            uint64_t csn = MOT::GetCSNManager().GetNextCSN();
            rc = htap::HTAPAdaptor::Commit(csn);
        } else if (txnState != MOT::TxnState::TXN_ROLLBACK && txnState != MOT::TxnState::TXN_COMMIT) {
            ereport(ERROR,
                (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), errmsg("commit prepared without prepare (%u)", txnState)));
        }
        if (rc != MOT::RC_OK) {
            MemoryEreportError();
            ereport(ERROR, (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                errmsg("could not serialize access due to concurrent update")));
        }
        txn->SetTxnState(MOT::TxnState::TXN_COMMIT);
    } else if (event == XACT_EVENT_ROLLBACK_PREPARED) {
        htap::HTAPAdaptor::RollbackPrepared();
        txn->SetTxnState(MOT::TxnState::TXN_ROLLBACK);
    }

    return HTAP_RC_OK;
}

RC HTAPInfra::HandleSubXactEvent(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid)
{
    MOT::TxnManager *txn = nullptr;
    MOT::RC rc = MOT::RC_OK;
    bool hasCommitedSubTxn = false;
    int savedInterruptHoldoffCount = t_thrd.int_cxt.InterruptHoldoffCount;
    PG_TRY();
    {
        txn = HTAPGetSafeTxn();
    }
    PG_CATCH();
    {
        /*
         * handle ereport error will reset InterruptHoldoffCount issue,
         * if not handle, caller may fail on assert
         */
        t_thrd.int_cxt.InterruptHoldoffCount = savedInterruptHoldoffCount;
        switch (event) {
            case SUBXACT_EVENT_ABORT_SUB:
                return HTAP_RC_OK;
            default:
                PG_RE_THROW();
        }
    }
    PG_END_TRY();

    switch (event) {
        case SUBXACT_EVENT_START_SUB:
            elog(DEBUG2, "Start sub transaction %lu, parent %lu", mySubid, parentSubid);
            if (txn->IsTxnAborted()) {
                raiseAbortTxnError();
            }
            txn->StartSubTransaction(mySubid, u_sess->utils_cxt.XactIsoLevel);
            break;
        case SUBXACT_EVENT_COMMIT_SUB:
            elog(DEBUG2, "Commit sub transaction %lu, parent %lu", mySubid, parentSubid);
            if (txn->IsTxnAborted() && txn->HasCommitedSubTxnDDL()) {
                raiseAbortTxnError();
            }
            txn->CommitSubTransaction(mySubid);
            break;
        case SUBXACT_EVENT_ABORT_SUB:
            elog(DEBUG2, "Abort sub transaction %lu, parent %lu", mySubid, parentSubid);
            rc = txn->RollbackSubTransaction(mySubid);
            hasCommitedSubTxn = txn->HasCommitedSubTxnDDL();
            if (rc != MOT::RC_OK) {
                htap::HTAPAdaptor::Rollback();
                txn->SetTxnAborted();
                if (hasCommitedSubTxn) {
                    txn->SetHasCommitedSubTxnDDL();
                }
            }
            break;
        default:
            break;
    }
    return HTAP_RC_OK;
}
} // namespace htap